/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.python;

import org.apache.flink.annotation.Experimental;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.configuration.description.Description;

/** Configuration options for the Python API. */
@PublicEvolving
public class PythonOptions {

    /** The maximum number of elements to include in a bundle. */
    public static final ConfigOption<Integer> MAX_BUNDLE_SIZE =
            ConfigOptions.key("python.fn-execution.bundle.size")
                    .defaultValue(100000)
                    .withDescription(
                            "The maximum number of elements to include in a bundle for Python "
                                    + "user-defined function execution. The elements are processed asynchronously. "
                                    + "One bundle of elements are processed before processing the next bundle of elements. "
                                    + "A larger value can improve the throughput, but at the cost of more memory usage and higher latency.");

    /** The maximum time to wait before finalising a bundle (in milliseconds). */
    public static final ConfigOption<Long> MAX_BUNDLE_TIME_MILLS =
            ConfigOptions.key("python.fn-execution.bundle.time")
                    .defaultValue(1000L)
                    .withDescription(
                            "Sets the waiting timeout(in milliseconds) before processing a bundle for "
                                    + "Python user-defined function execution. The timeout defines how long the elements of a bundle will be "
                                    + "buffered before being processed. Lower timeouts lead to lower tail latencies, but may affect throughput.");

    /** The maximum number of elements to include in an arrow batch. */
    public static final ConfigOption<Integer> MAX_ARROW_BATCH_SIZE =
            ConfigOptions.key("python.fn-execution.arrow.batch.size")
                    .defaultValue(10000)
                    .withDescription(
                            "The maximum number of elements to include in an arrow batch for Python "
                                    + "user-defined function execution. The arrow batch size should not exceed the "
                                    + "bundle size. Otherwise, the bundle size will be used as the arrow batch size.");

    /** The configuration to enable or disable metric for Python execution. */
    public static final ConfigOption<Boolean> PYTHON_METRIC_ENABLED =
            ConfigOptions.key("python.metric.enabled")
                    .defaultValue(true)
                    .withDescription(
                            "When it is false, metric for Python will be disabled. You can "
                                    + "disable the metric to achieve better performance at some circumstance.");

    public static final ConfigOption<String> PYTHON_FILES =
            ConfigOptions.key("python.files")
                    .stringType()
                    .noDefaultValue()
                    .withDescription(
                            "Attach custom python files for job. The standard python resource file suffixes such as .py/.egg/.zip or "
                                    + "directory are all supported. These files will be added to the PYTHONPATH of both the local "
                                    + "client and the remote python UDF worker. Files suffixed with .zip will be extracted and added to PYTHONPATH. "
                                    + "Comma (',') could be used as the separator to specify multiple files. The option is equivalent to the command line option \"-pyfs\". ");

    public static final ConfigOption<String> PYTHON_REQUIREMENTS =
            ConfigOptions.key("python.requirements")
                    .stringType()
                    .noDefaultValue()
                    .withDescription(
                            "Specify a requirements.txt file which defines the third-party "
                                    + "dependencies. These dependencies will be installed and added to the PYTHONPATH of "
                                    + "the python UDF worker. A directory which contains the installation packages of "
                                    + "these dependencies could be specified optionally. Use '#' as the separator if the "
                                    + "optional parameter exists. The option is equivalent to the command line option "
                                    + "\"-pyreq\".");

    public static final ConfigOption<String> PYTHON_ARCHIVES =
            ConfigOptions.key("python.archives")
                    .stringType()
                    .noDefaultValue()
                    .withDescription(
                            "Add python archive files for job. The archive files will be extracted "
                                    + "to the working directory of python UDF worker. Currently only zip-format is "
                                    + "supported. For each archive file, a target directory is specified. If the target "
                                    + "directory name is specified, the archive file will be extracted to a name can "
                                    + "directory with the specified name. Otherwise, the archive file will be extracted to "
                                    + "a directory with the same name of the archive file. The files uploaded via this "
                                    + "option are accessible via relative path. '#' could be used as the separator of the "
                                    + "archive file path and the target directory name. Comma (',') could be used as the "
                                    + "separator to specify multiple archive files. This option can be used to upload the "
                                    + "virtual environment, the data files used in Python UDF. The data files could be "
                                    + "accessed in Python UDF, e.g.: f = open('data/data.txt', 'r'). The option is "
                                    + "equivalent to the command line option \"-pyarch\".");

    public static final ConfigOption<String> PYTHON_EXECUTABLE =
            ConfigOptions.key("python.executable")
                    .stringType()
                    .defaultValue("python")
                    .withDescription(
                            "Specify the path of the python interpreter used to execute the python "
                                    + "UDF worker. The python UDF worker depends on Python 3.5+, Apache Beam "
                                    + "(version == 2.23.0), Pip (version >= 7.1.0) and SetupTools (version >= 37.0.0). "
                                    + "Please ensure that the specified environment meets the above requirements. The "
                                    + "option is equivalent to the command line option \"-pyexec\".");

    public static final ConfigOption<String> PYTHON_CLIENT_EXECUTABLE =
            ConfigOptions.key("python.client.executable")
                    .stringType()
                    .defaultValue("python")
                    .withDescription(
                            Description.builder()
                                    .text(
                                            "The path of the Python interpreter used to launch the Python process when submitting the "
                                                    + "Python jobs via \"flink run\" or compiling the Java/Scala jobs containing Python UDFs. "
                                                    + "Equivalent to the environment variable PYFLINK_CLIENT_EXECUTABLE. "
                                                    + "The priority is as following: ")
                                    .linebreak()
                                    .text(
                                            "1. the configuration 'python.client.executable' defined in the source code;")
                                    .linebreak()
                                    .text("2. the environment variable PYFLINK_CLIENT_EXECUTABLE;")
                                    .linebreak()
                                    .text(
                                            "3. the configuration 'python.client.executable' defined in flink-conf.yaml")
                                    .build());

    /** Whether the memory used by the Python framework is managed memory. */
    public static final ConfigOption<Boolean> USE_MANAGED_MEMORY =
            ConfigOptions.key("python.fn-execution.memory.managed")
                    .defaultValue(true)
                    .withDescription(
                            String.format(
                                    "If set, the Python worker will configure itself to use the "
                                            + "managed memory budget of the task slot. Otherwise, it will use the Off-Heap Memory "
                                            + "of the task slot. In this case, users should set the Task Off-Heap Memory using the "
                                            + "configuration key %s.",
                                    TaskManagerOptions.TASK_OFF_HEAP_MEMORY.key()));

    /** The maximum number of states cached in a Python UDF worker. */
    @Experimental
    public static final ConfigOption<Integer> STATE_CACHE_SIZE =
            ConfigOptions.key("python.state.cache-size")
                    .defaultValue(1000)
                    .withDescription(
                            "The maximum number of states cached in a Python UDF worker. Note that this "
                                    + "is an experimental flag and might not be available in future releases.");

    /** The maximum number of cached items which read from Java side in a Python MapState. */
    @Experimental
    public static final ConfigOption<Integer> MAP_STATE_READ_CACHE_SIZE =
            ConfigOptions.key("python.map-state.read-cache-size")
                    .defaultValue(1000)
                    .withDescription(
                            "The maximum number of cached entries for a single Python MapState. "
                                    + "Note that this is an experimental flag and might not be available in future releases.");

    /** The maximum number of write requests cached in a Python MapState. */
    @Experimental
    public static final ConfigOption<Integer> MAP_STATE_WRITE_CACHE_SIZE =
            ConfigOptions.key("python.map-state.write-cache-size")
                    .defaultValue(1000)
                    .withDescription(
                            "The maximum number of cached write requests for a single Python "
                                    + "MapState. The write requests will be flushed to the state backend (managed in "
                                    + "the Java operator) when the number of cached write requests exceed this limit. "
                                    + "Note that this is an experimental flag and might not be available in future "
                                    + "releases.");

    /**
     * The maximum number of entries sent to Python UDF worker per request when iterating a Python
     * MapState.
     */
    @Experimental
    public static final ConfigOption<Integer> MAP_STATE_ITERATE_RESPONSE_BATCH_SIZE =
            ConfigOptions.key("python.map-state.iterate-response-batch-size")
                    .defaultValue(1000)
                    .withDescription(
                            "The maximum number of the MapState keys/entries sent to Python UDF worker "
                                    + "in each batch when iterating a Python MapState. Note that this is an experimental flag "
                                    + "and might not be available "
                                    + "in future releases.");
}
